package com.jxtest.studykafka.kafkatest.kafka;
import com.jxtest.studykafka.kafkatest.constant.KafkaConst;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @Author 18011618
 * @Description
 * @Date 14:50 2018/7/20
 * @Modify By
 */
@Component
@Slf4j
public class KafkaConsumer {

    @KafkaListener(topics = {KafkaConst.CONSUMER_TOPIC},groupId = "test")
    public void consumer(ConsumerRecord<?,?> consumerRecord){
        try {
            //判断是否为null
            Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
            log.info("收到消息topic:{},内容：{} ",KafkaConst.CONSUMER_TOPIC, kafkaMessage);
            if(kafkaMessage.isPresent()){
                //得到Optional实例中的值
                Object message = kafkaMessage.get();
                log.info("消费消息topic:{},内容:{}",KafkaConst.CONSUMER_TOPIC,message);
            }
        }catch (Exception e){
            log.error("消费kaka消息失败");
        }

    }
}
